package com.atbeijing.D02;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 键控流转换算子keyBy
 * 可以将流分成多个
 */
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<SensorReading> stream = env.addSource(new SensorSource());

        //所有数据都分到一个流
        KeyedStream<SensorReading, Boolean> keyedStream1 = stream.keyBy(new KeySelector<SensorReading, Boolean>() {
            @Override
            public Boolean getKey(SensorReading sensorReading) throws Exception {
                return true;
            }
        });

        //将数据分成两个流,第一个参数:输入数据类型为SensorReading,第二个参数:分流表达式的返回类型,true和false
        KeyedStream<SensorReading, Boolean> keyedStream2 = stream.keyBy(new KeySelector<SensorReading, Boolean>() {
            @Override
            public Boolean getKey(SensorReading sensorReading) throws Exception {
                return sensorReading.temperature% 2 == 0;
            }

        });
        keyedStream2.print();

        //分成三个流0,1,2
        DataStreamSource<Integer> stream1 = env.fromElements(1, 2, 3, 4, 5, 6, 7);
        KeyedStream<Integer, Integer> keyedStream3 = stream1.keyBy(new KeySelector<Integer, Integer>() {
            @Override
            public Integer getKey(Integer integer) throws Exception {
                return integer % 3;
            }
        });

        // 使用HashMap来做逻辑分区
//        HashMap<String, ArrayList<SensorReading>> hashMap = new HashMap<>();



        env.execute();
    }
}
